package com.maoshi.shop.seckill.api.mq.redis;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.maoshi.shop.common.response.ServerResponseEntity;
import com.maoshi.shop.common.util.Json;
import com.maoshi.shop.seckill.api.socket.WebSocketServer;
import com.maoshi.shop.seckill.comment.dto.SeckillOrderMergerDto;
import com.maoshi.shop.seckill.comment.service.SeckillOrderService;
import lombok.AllArgsConstructor;
import org.redisson.Redisson;
import org.redisson.api.RBlockingDeque;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.util.concurrent.*;

/**
 * 秒杀订单队列监听 使用BlockingDeque 保证一台机器只能收到一条消息
 * @author LGH
 */
@Component
@AllArgsConstructor
public class SeckillOrderDequeListener implements ApplicationListener<ContextRefreshedEvent> {

    private final Redisson redisson;

    private final SeckillOrderService seckillOrderService;
    // 秒杀的消息队列要是单线程，那么这个秒杀还要不要啦，当然是不能单线程咯
    private static Executor executor = new ThreadPoolExecutor(
            5,
            20,
            5L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(),
            new ThreadFactoryBuilder()
                    .setNameFormat("seckill-redis-listener-pool-%d").build());



    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {

        RBlockingDeque<SeckillOrderMergerDto> queue = redisson.getBlockingDeque("seckill");
        Thread dequeThread = new Thread(() -> {
            while (true) {
                final SeckillOrderMergerDto dto;
                try {
                    dto = queue.take();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    continue;
                }
                try {
                    executor.execute(() -> {
                        String orderNumber;

                        try {
                            orderNumber = seckillOrderService.submit(dto);
                        } catch (Exception e) {
                            // 推送前台秒杀失败的消息
                            WebSocketServer.sendMessage(dto.getMsgId(), Json.toJsonString(ServerResponseEntity.fail(e.getMessage())));
                            return;
                        }
                        // 秒杀还是不要移除为妙
                        // seckillSkuService.removeSeckillCacheSkuById(shopCartItem.getSeckillSkuId(),shopCartItem.getSeckillId());

                        // 推送前台秒杀成功的消息
                        WebSocketServer.sendMessage(dto.getMsgId(), Json.toJsonString(ServerResponseEntity.success(orderNumber)));
                    });
                } catch (Exception e) {
                    // 推送前台秒杀失败的消息
                    WebSocketServer.sendMessage(dto.getMsgId(), Json.toJsonString(ServerResponseEntity.fail(e.getMessage())));
                }


            }
        });

        dequeThread.setDaemon(true);

        dequeThread.start();
    }

}
